package com.zyx.pulsardemo.basic.consumers;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;


/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 17:53
 * desc: 异步处理消息案例
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class AsyncConsumerDemo {

    private static final LinkedBlockingQueue<CompletableFuture<Message<String>>> OUTSTANDING_MESSAGES =
            new LinkedBlockingQueue<CompletableFuture<Message<String>>>(1000);

    public static void main(String[] args) throws PulsarClientException, InterruptedException, ExecutionException {

        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl("pulsar://broker:6650")
                .build();

        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
                .subscriptionName("pulsarSubscriptionName")
                .topic("pulsarTopicName")
                .ackTimeout(240, TimeUnit.SECONDS)
                .subscriptionType(SubscriptionType.Exclusive)
                .subscribe();

        new Thread(() -> {
            CompletableFuture<Message<String>> future;

            while ((future = consumer.receiveAsync()) != null) {
                OUTSTANDING_MESSAGES.add(future);
            }
        }).start();

        new Thread(() -> {
            while(true) {
                try {
                    CompletableFuture<Message<String>> future = OUTSTANDING_MESSAGES.take();
                    Message<String> msg = future.get();

                    // Process the message
                    consumer.acknowledgeAsync(msg.getMessageId());

                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }


}
